package org.aplikator.server.persistence.search;

import com.typesafe.config.Config;
import org.aplikator.client.shared.data.PrimaryKey;
import org.aplikator.client.shared.data.Record;
import org.aplikator.client.shared.data.SearchResult;
import org.aplikator.client.shared.descriptor.EntityDTO;
import org.aplikator.server.Context;
import org.aplikator.server.DescriptorRegistry;
import org.aplikator.server.descriptor.Entity;
import org.aplikator.server.descriptor.Property;
import org.aplikator.server.descriptor.View;
import org.aplikator.server.persistence.Persister;
import org.aplikator.server.persistence.PersisterFactory;
import org.aplikator.server.persistence.PersisterTriggers;
import org.aplikator.server.persistence.Transaction;
import org.aplikator.server.util.Configurator;
import org.elasticsearch.action.get.GetRequestBuilder;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.FieldQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.jboss.errai.marshalling.client.Marshalling;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @author eskymo
 * 
 *         ElasticSearch implements aplikators interface Search - it's an
 *         implementation for ElasticSearch engine
 * 
 */
public class ElasticSearch implements Search {

	private Client client;
	private static final Logger LOG = Logger.getLogger(ElasticSearch.class
			.getName());
	private static final String SEARCHINDEX = "aplikator.searchIndexName";
	private static final String SEARCHHOST = "aplikator.searchHost";
	private static final String SEARCHPORT = "aplikator.searchPort";
	private static final String SEARCHCLUSTERNAME = "aplikator.searchClusterName";
	private static final Integer DEFAULT_TRAVERSE_LEVEL = 1;
	private static final Boolean DEFAULT_INCLUDE_COLLECTIONS = true;

	private Persister persister;
	private Transaction tx;
	private String indexName;

	/**
	 * Constructor gets the index and search engine parameters and establishes
	 * the connection to the search engine
	 */
	protected ElasticSearch() {
		Config config = Configurator.get().getConfig();
		if (config.getString(SEARCHCLUSTERNAME)==null) {
			LOG.severe("Property "+SEARCHCLUSTERNAME+" not set !");
			throw new IllegalStateException("Property "+SEARCHCLUSTERNAME+" not set !");
			
		}
		if (config.getString(SEARCHHOST)==null) {
			LOG.severe("Property "+SEARCHHOST+" not set !");
			throw new IllegalStateException("Property "+SEARCHHOST+" not set !");
			
		}
		if (config.getString(SEARCHPORT)==null) {
			LOG.severe("Property "+SEARCHPORT+" not set !");
			throw new IllegalStateException("Property "+SEARCHPORT+" not set !");			
		}
		if (config.getString(SEARCHINDEX)==null) {
			LOG.severe("Property "+SEARCHINDEX+" not set !");
			throw new IllegalStateException("Property "+SEARCHINDEX+" not set !");
			
		}
		this.indexName = config.getString(SEARCHINDEX);
		this.client = getClient(config.getString(SEARCHCLUSTERNAME),
				config.getString(SEARCHHOST), config.getInt(SEARCHPORT));
		this.persister = PersisterFactory.getPersister();
	}
	
	@SuppressWarnings("rawtypes")
	private Record getReferencedRecord(Record rec, Property p) {
		Record refRecord;
		if (p.getRefferedThrough()!=null) {
			 refRecord = getReferencedRecord(rec, p.getRefferedThrough());
			 if (refRecord.getValue(p.getId()) instanceof Record) {
				 refRecord = (Record) refRecord.getValue(p.getId());
			 } 
		} else {
			refRecord = (Record) rec.getValue(p.getId());
		}
		return refRecord;		
	}

	/**
	 * 
	 * Generates a Record from class CompleteRecord using the localization and
	 * triggers provided from context
	 * 
	 * @param view
	 * @param completeRecord
	 * @param ctx
	 * @return
	 */
	@SuppressWarnings({ "rawtypes", "unchecked" })
	private Record generateRecord(View view, Record completeRecord, Context ctx) {
		PersisterTriggers trigger = view.getPersisterTriggers();

		// Creates new Record instance from PrimaryKey
		Record rec = new Record(new PrimaryKey(completeRecord.getPrimaryKey()
				.getEntityId(), completeRecord.getPrimaryKey().getId()));

		// insert values into the Record based on View setup
		for (int i = 0; i < view.getProperties().size(); i++) {
			Property p = view.getProperties().get(i);

			// if a property is a Reference then the value must be get from
			// referencedRecord list from CompleteRecord instance
			if (p.getRefferedThrough() != null) {
//				Record refRecord = (Record) completeRecord.getValue(p
//						.getRefferedThrough().getId());
				
				Record refRecord = getReferencedRecord(completeRecord, p);
//				Record refRecord = (Record) completeRecord.getValue(p
//						.getRefferedThrough().getId());
				p.setValue(rec, (Serializable) refRecord.getValue(p.getId()));
			} else {
				p.setValue(rec,
						(Serializable) completeRecord.getValue(p.getId()));
			}
		}
		if (trigger != null) {
			trigger.onLoad(rec, ctx);
		}
		rec.resetDirtyFlags();
		return rec;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.aplikator.server.persistence.search.Search#getPagefromSearch(org.
	 * aplikator.client.shared.descriptor.ViewDTO, java.lang.String, int, int,
	 * org.aplikator.server.Context)
	 */
	@Override
	public SearchResult getPagefromSearch(String vdId, String searchArgument,
			int pageOffset, int pageSize, org.aplikator.server.Context ctx) {

		if (vdId == null) {
			return getPagefromSearch(searchArgument, pageOffset, pageSize, ctx);
		}

		View view = (View) DescriptorRegistry.get().getDescriptionItem(vdId);
		// if view is not supplied, call non-typed search accross all entities

		SearchResult completeRecords = search(searchArgument, view.getEntity()
				.getId(), pageOffset, pageSize);

		List<Record> retval = new ArrayList<Record>();
		try {
			int recCounter = 0;
			for (Record completeRecord : completeRecords.getRecords()) {
				Record rec = generateRecord(view, completeRecord, ctx);
				retval.add(rec);
				recCounter++;
				if (pageSize > 0 && recCounter >= pageSize) {
					break;
				}

			}
		} catch (Throwable th) {
			LOG.log(Level.SEVERE, "Error in GETPAGEfromSearch", th);
		}
		return new SearchResult(retval, completeRecords.getCount());
	}

	/**
	 * 
	 * creates an instance of a client for a search engine
	 * 
	 * @param clusterName
	 * @param clusterHost
	 * @param clusterPort
	 * @return
	 */
	private Client getClient(String clusterName, String clusterHost,
			int clusterPort) {
		if (this.client == null) {
			Settings settings = ImmutableSettings.settingsBuilder()
					.put("cluster.name", clusterName).build();

			Client client = new TransportClient(settings)
					.addTransportAddress(new InetSocketTransportAddress(
							clusterHost, clusterPort));
			this.client = client;
		}
		return client;
	}

	@Override
	public void index(Record record) {
		String type = record.getPrimaryKey().getEntityId();
		IndexRequest indexRequest = new IndexRequest(indexName, type);
		String json =  Marshalling.toJSON(record);
		indexRequest.source(json);
		IndexResponse indexResponse = client.index(indexRequest).actionGet();
		LOG.fine("Indexed into " + indexResponse.getId());
	}

	@Override
	public void index(EntityDTO entityDTO) {
		Entity entity = (Entity) DescriptorRegistry.get().getDescriptionItem(
				entityDTO.getId());
		try {
			client.prepareDeleteByQuery(indexName)
					.setQuery(QueryBuilders.matchAllQuery())
					.setTypes(entityDTO.getId()).execute().get();
		} catch (Exception e) {
			LOG.severe("Problem in delete type:" + e.getMessage());
			throw new IllegalStateException(e);
		}
		List<PrimaryKey> PKs = persister.getEntityPKs(entity, null, null);
		for (PrimaryKey primaryKey : PKs) {
			Record record = persister.getCompleteRecord(primaryKey,
					DEFAULT_TRAVERSE_LEVEL, DEFAULT_INCLUDE_COLLECTIONS,tx);
			index(record);
		}
	}

	@Override
	public SearchResult search(String searchArgument, String type, int offset,
			int size) {
		List<Record> records = new ArrayList<Record>();
		SearchResponse response = null;
		SearchRequestBuilder searchBuilder = client.prepareSearch(indexName)
				.setQuery(QueryBuilders.queryString(searchArgument));
		if (type != null) {
			searchBuilder.setTypes(type);
		}
		if (offset > 0) {
			searchBuilder.setFrom(offset);
		}
		if (size > 0) {
			searchBuilder.setSize(size);
		}
		try {
			response = searchBuilder.execute().get();

		} catch (Exception e) {
			LOG.severe("Problem in search:" + e.getMessage());
			throw new IllegalStateException(e);
		}

		SearchHits hits = response.getHits();
		Iterator<SearchHit> hitIterator = hits.iterator();
		while (hitIterator.hasNext()) {
			SearchHit hit = hitIterator.next();
			Record record = (Record) Marshalling.fromJSON(hit
					.getSourceAsString(), Record.class);
			records.add(record);
		}
		return new SearchResult(records, hits.totalHits());
	}

	private List<String> searchIDs(String searchArgument) {
		List<String> IDs = new ArrayList<String>();
		SearchResponse response = null;
		FieldQueryBuilder builder = QueryBuilders.fieldQuery("*primaryKey",searchArgument);
		builder.autoGeneratePhraseQueries(true);
		builder.analyzer("standard");
		
		SearchRequestBuilder searchBuilder = client.prepareSearch(indexName)
				.setQuery(builder);
		try {
			response = searchBuilder.execute().get();

		} catch (Exception e) {
			LOG.severe("Problem in search:" + e.getMessage());
			throw new IllegalStateException(e);
		}

		SearchHits hits = response.getHits();
		Iterator<SearchHit> hitIterator = hits.iterator();
		while (hitIterator.hasNext()) {
			SearchHit hit = hitIterator.next();
			IDs.add(hit.getId());
		}
		
		
		return IDs;
	}

	@Override
	public SearchResult getPagefromSearch(String searchArgument,
			int pageOffset, int pageSize, Context ctx) {
		ArrayList<Record> records = new ArrayList<Record>();
		SearchResult completeRecords = search(searchArgument, pageOffset,
				pageSize);
		for (Record completeRecord : completeRecords.getRecords()) {
			// TODO - missing info from entity to create record preview
			Entity entity = (Entity) DescriptorRegistry.get()
					.getDescriptionItem(
							completeRecord.getPrimaryKey().getEntityId());
			View view = entity.view();
			Record rec = generateRecord(view, completeRecord, ctx);
			records.add(rec);
		}
		return new SearchResult(records, completeRecords.getCount());
	}

	@Override
	public SearchResult search(String searchArgument, int offset, int size) {
		return search(searchArgument, null, offset, size);
	}

	@Override
	public void update(PrimaryKey primaryKey) {
		String primaryKeyString = primaryKey.getSerializationString();
		Set<PrimaryKey> processedPKs  = new HashSet<PrimaryKey>();
		List<String> listIDs = searchIDs(primaryKeyString);
		if (listIDs.size()==0) {
			Record newRecord = persister.getCompleteRecord(
					primaryKey, DEFAULT_TRAVERSE_LEVEL,
					DEFAULT_INCLUDE_COLLECTIONS,tx);
			index(newRecord);
			return;
		}
		for (String updatedID : listIDs) {
			GetRequestBuilder getRequest = new GetRequestBuilder(client,
					indexName);
			getRequest.setId(updatedID);
			GetResponse getResponse = client.get(getRequest.request())
					.actionGet();
			Record recordFound = (Record) Marshalling.fromJSON(getResponse
					.getSourceAsString(),Record.class);
			client.prepareDelete(indexName, recordFound.getPrimaryKey().getEntityId(),updatedID).execute().actionGet();
			if (processedPKs.contains(recordFound.getPrimaryKey())) {
			    // just in case record is indexed twice by mistake
			    continue;
			}
			Record newRecord = persister.getCompleteRecord(
					recordFound.getPrimaryKey(), DEFAULT_TRAVERSE_LEVEL,
					DEFAULT_INCLUDE_COLLECTIONS,tx);
			index(newRecord);
			processedPKs.add(recordFound.getPrimaryKey());
		}
	}

	@Override
	public void insert(PrimaryKey primaryKey) {
		Record newRecord = persister.getCompleteRecord(primaryKey,
				DEFAULT_TRAVERSE_LEVEL, DEFAULT_INCLUDE_COLLECTIONS,tx);
		index(newRecord);
	}

	@Override
	public void delete(PrimaryKey primaryKey) {
		String primaryKeyString = primaryKey.getSerializationString();
		List<String> listIDs = searchIDs(primaryKeyString);
		for (String updatedID : listIDs) {
			GetRequestBuilder getRequest = new GetRequestBuilder(client,
					indexName);
			getRequest.setId(updatedID);
			client.prepareDelete().setId(updatedID).execute().actionGet();
		}
	}

    @Override
    public void init(Transaction tx) {
        if (tx!=null) {
            this.tx=tx;
        }       
    }

    @Override
    public void finish() {
        // TODO Auto-generated method stub
        
    }
}